using System;
using System.Collections.Generic;
using log4net;
using Needle.Bus.Messages;
using Needle.Bus.Messages.Management;
using System.Diagnostics;

namespace Needle.Bus.Engine.Schedulers
{
    public class StagePump : IWorkItem
    {
        private static ILog _log = LogManager.GetLogger(typeof(StagePump));
     
        private Watch _watch;
        private IStage _stage;
  
        public IStage Stage
        {
            get { return _stage; }
        }

        public StagePump(IStage stage)
        {
            if (stage == null)
                throw new ArgumentNullException("stage", "stage is null.");
            _stage = stage;
            _watch = new Watch();
        }

        public void Run()
        {
            if (_stage.Activate())
            {
                try
                {
                    using (var ctx = (new Bus(Bus.Context.Engine, _stage)).SetCurrent())
                    {
                        if (_stage.BatchMode.Enabled)
                        {
                            PumpMessagesBatched();
                        }
                        else
                        {
                            PumpMessages();
                        }
                    }
                }
                catch (Exception ex)
                {
                    _log.Error("Error while processing messages in stage " + _stage.Id, ex);
                }
                finally
                {
                    _stage.Deactivate();
                }
            }
        }

        private void PumpMessages()
        {
            var buffer = _stage.Buffer;
            var filter = _stage.Filter;
            var handler = _stage.Handler;
            var perfCounter = _stage.PerfCounter;

            IMessage message = null;
            while (buffer.TryDequeue(out message))
            {
                _watch.Restart();
                handler.Process(message);
                perfCounter.Record(1, Stopwatch.GetTimestamp(), _watch.Elapsed.Ticks);
                filter.Update();
            }
        }

        private void PumpMessagesBatched()
        {
            var buffer = _stage.Buffer;
            var filter = _stage.Filter;
            var handler = _stage.Handler;
            var perfCounter = _stage.PerfCounter;
            var minBatchSize = _stage.BatchMode.MinBatchSize;
            var maxBatchSize = _stage.BatchMode.MaxBatchSize;

            IList<IMessage> messages = null;
            while (buffer.Count >= minBatchSize)
            {   /* race condition here that could cause less 
                 * messages to be dequeued than minBatchSize */
                if (buffer.TryDequeue(maxBatchSize, out messages))
                {
                    _watch.Restart();
                    handler.Process(messages);
                    perfCounter.Record(messages.Count, Stopwatch.GetTimestamp(), _watch.Elapsed.Ticks);
                    filter.Update();
                }
                else
                {
                    break;
                }
            }
        }

        #region Watch

        private class Watch : Stopwatch
        {
            public void Restart()
            {
                Reset();
                Start();
            }
        }

        #endregion

    }
}

